package RegionPartition;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class RegionPartitionMapper extends Mapper<LongWritable, Text, Text, RegionPartitionFlowBean> {
    // 定义一个TreeMap作为存储数据的容器（按key排序）-=======按key排序是不是导致了结果输出中只有10行？？？？？
//    private TreeMap<RegionPartitionFlowBean, Text> flowMap = new TreeMap<>();

    @Override
    protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
        // 1. 获取一行
        String line = value.toString();

        // 2. 切割
        String[] fields = line.split(",");

        // 3. 封装数据
        String id = fields[0];
//        int loadAmnt = Integer.parseInt(fields[1]);
//        int term = Integer.parseInt(fields[2]);
//        float interestRate = Float.parseFloat(fields[3]);
//        float installment = Float.parseFloat(fields[4]);
//        String grade = fields[5];
//        String subGrade = fields[6];
//        int employmentTitle = Integer.parseInt(fields[7]);
//        String employmentLength = fields[8];
//        int homeOwnership = Integer.parseInt(fields[9]);
//        int annualIncome = Integer.parseInt(fields[10]);
//        int verificationStatus = Integer.parseInt(fields[11]);
//        String issueDate = fields[12];
//        int purpose = Integer.parseInt(fields[13]);
//        int postCode = Integer.parseInt(fields[14]);
        int regionCode = Integer.parseInt(fields[15]);
//        float dti = Float.parseFloat(fields[16]);
//        int delinquency_2years = Integer.parseInt(fields[17]);
//        int ficoRangeLow = Integer.parseInt(fields[18]);
//        int ficoRangeHigh = Integer.parseInt(fields[19]);
//        int openAcc = Integer.parseInt(fields[20]);
//        int pubRec = Integer.parseInt(fields[21]);
//        int pubRecBankruptcies = Integer.parseInt(fields[22]);
//        int revolBal = Integer.parseInt(fields[23]);
//        float revolUtil = Float.parseFloat(fields[24]);
//        int totalAcc = Integer.parseInt(fields[25]);
//        int initialListStatus = Integer.parseInt(fields[26]);
//        int applicationType = Integer.parseInt(fields[27]);
//        String earliesCreditLine = fields[28];
//        int title = Integer.parseInt(fields[29]);
//        int policyCode = Integer.parseInt(fields[30]);

        Text idtext = new Text(id);
        RegionPartitionFlowBean flowBean = new RegionPartitionFlowBean(regionCode);

        // 4. 向TreeMap中添加数据
//        flowMap.put(flowBean, idtext);
        context.write(idtext, flowBean);
    }

    /*@Override
    protected void cleanup(Mapper.Context context) throws IOException, InterruptedException {
        // 6 遍历treeMap集合，输出数据
        Iterator<RegionPartitionFlowBean> iterator = flowMap.keySet().iterator();

        while (iterator.hasNext()) {
            RegionPartitionFlowBean valueOut = iterator.next();
            Text keyOut = flowMap.get(valueOut);

            context.write(keyOut, valueOut);
        }
    }*/
}
